package org.fastsyncer.manager.increment.logmnr;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.support.rowset.SqlRowSet;
import org.springframework.jdbc.support.rowset.SqlRowSetMetaData;

public class LogmnrUtil {

	/** 数据同步最大SCN号 */
	private static final Integer SCN_MAX_POS = 1000;

	/**
	 * 查询最后一次记录点
	 * 
	 * @param jdbcTemplate
	 * @param dictionary
	 *            字典文件路径
	 * @param scnStart
	 *            记录点
	 * @return 最近记录点
	 * @throws SQLException
	 */
	public static Object parseMaxScn(JdbcTemplate jdbcTemplate, String dictionary, String scnStart)
			throws SQLException {
		// 默认只查询正在运行的日志
		String loadLogs = loadLogs(jdbcTemplate, false);
		jdbcTemplate.execute(loadLogs);

		// 如果是第一次获取增量，就去查询Logminer最后一次记录点
		String sql = "BEGIN dbms_logmnr.start_logmnr(dictfilename=>'" + dictionary
				+ "',OPTIONS =>DBMS_LOGMNR.COMMITTED_DATA_ONLY); END;";
		jdbcTemplate.execute(sql);

		sql = "SELECT MAX(SCN) FROM V$LOGMNR_CONTENTS T";
		SqlRowSet rs = jdbcTemplate.queryForRowSet(sql);
		Object scn = 0;
		while (rs.next()) {
			scn = rs.getObject(1);
		}
		return scn;
	}

	/**
	 * 解析Oracle增量日志，一次最多可返回1000条
	 * 
	 * @param jdbc
	 *            JdbcTemplate连接实例
	 * @param table
	 *            扫描表
	 * @param scanAll
	 *            扫面所有日志,默认false只扫描正在运行的日志
	 * @param dictionary
	 *            扫描日志字典
	 * @param scnStart
	 *            扫描日志的偏移开始位置
	 * @return 增量日志（只返回增改删操作类型的日志）
	 * @throws SQLException
	 * @throws JSONException
	 */
	public static JSONObject parse(JdbcTemplate jdbcTemplate, String scanTable, boolean scanAll, String dictionary,
			String scnStart) throws SQLException, JSONException {
		// 启动logmnr
		startLogmnr(jdbcTemplate, dictionary, scnStart, scanAll);

		// 扫描日志
		JSONObject parseLogs = parseLogs(jdbcTemplate, scanTable);
		return parseLogs;
	}

	private static String loadLogs(JdbcTemplate jdbcTemplate, boolean scanAll) throws SQLException {

		// 只查询正在运行的日志
		String sql = "SELECT MEMBER FROM V$LOGFILE";
		// 默认解析当前正在运行的日志
		if (!scanAll) {
			sql = "SELECT S.MEMBER FROM V$LOG L,V$LOGFILE S WHERE L.GROUP# = S.GROUP# AND L.STATUS = 'CURRENT'";
		}

		// 打印获分析日志文件信息
		StringBuffer runningLogSQL = new StringBuffer();
		runningLogSQL.append("BEGIN");
		SqlRowSet queryForRowSet = jdbcTemplate.queryForRowSet(sql);
		int i = 0;
		while (queryForRowSet.next()) {
			i++;
			if (i == 1) {
				runningLogSQL.append(" dbms_logmnr.add_logfile(logfilename=>'" + queryForRowSet.getObject(1)
						+ "', options=>dbms_logmnr.NEW);");
				continue;
			}
			runningLogSQL.append(" dbms_logmnr.add_logfile(logfilename=>'" + queryForRowSet.getObject(1)
					+ "', options=>dbms_logmnr.ADDFILE);");
		}
		runningLogSQL.append(" END;");
		return runningLogSQL + "";
	}

	private static void startLogmnr(JdbcTemplate jdbcTemplate, String dictionary, String scnStartPos, boolean scanAll)
			throws SQLException {
		// 默认只查询正在运行的日志
		String loadLogs = loadLogs(jdbcTemplate, scanAll);
		jdbcTemplate.execute(loadLogs);

		// 启动解析器 StartScn=>'1326266',EndScn =>'1326536'
		String sql = "BEGIN dbms_logmnr.start_logmnr(StartScn=>'" + scnStartPos + "',dictfilename=>'" + dictionary
				+ "',OPTIONS =>DBMS_LOGMNR.COMMITTED_DATA_ONLY); END;";
		jdbcTemplate.execute(sql);
	}

	public static JSONObject parseLogs(JdbcTemplate jdbcTemplate, String scanTable) throws SQLException, JSONException {
		JSONObject result = new JSONObject();
		JSONArray msg = null;
		String sql = "SELECT SCN,OPERATION,SQL_REDO FROM V$LOGMNR_CONTENTS T WHERE T.SEG_NAME='" + scanTable + "'";
		SqlRowSet rs = jdbcTemplate.queryForRowSet(sql);
		String opr = null;
		String sqlRedo = null;

		int row = 0;
		Object scn = 0;
		while (rs.next()) {
			// 默认支持最大日志条数
			if (row == SCN_MAX_POS) {
				break;
			}
			row++;
			if (msg == null) {
				msg = new JSONArray();
			}
			scn = rs.getObject(1);
			// 过滤非INSER,UPDATE,DELETE日志
			opr = (String) rs.getObject(2);
			if (!opr.equalsIgnoreCase("INSERT") && !opr.equalsIgnoreCase("UPDATE") && !opr.equalsIgnoreCase("DELETE")) {
				continue;
			}
			// 获取执行SQL
			sqlRedo = (String) rs.getObject(3);

			// 解析日志
			try {
				JSONObject parseDML = parseDML(jdbcTemplate, opr, sqlRedo);
				msg.put(parseDML);
				// logger.info("解析后:" + parseDML + "\r");
			} catch (JSONException e) {
				e.printStackTrace();
			}
		}
		result.put("msg", msg);
		result.put("scn", scn);

		// 关闭Logmnr
		String closeSql = "BEGIN dbms_logmnr.end_logmnr; END;";
		jdbcTemplate.execute(closeSql);
		return result;
	}

	private static JSONObject parseDML(JdbcTemplate jdbcTemplate, String oper, String sql) throws JSONException {
		JSONObject json = null;
		switch (oper) {
		case LogmnrConstants.INSERT:
			json = insert(jdbcTemplate, oper, sql);
			break;
		case LogmnrConstants.UPDATE:
			json = update(jdbcTemplate, oper, sql);
			break;
		case LogmnrConstants.DELETE:
			json = delete(jdbcTemplate, oper, sql);
			break;
		default:
			break;
		}
		return json;
	}

	private static JSONObject insert(JdbcTemplate jdbcTemplate, String oper, String sql) throws JSONException {
		JSONObject insert = new JSONObject();
		JSONArray before = new JSONArray();
		JSONArray after = new JSONArray();

		// 截取列名字符串
		int startColIndex = sql.indexOf("(");
		int endColIndex = sql.indexOf(")");
		String sqlCol = sql.substring(startColIndex, endColIndex);
		// 找出列名
		String regex = "\"([\\s\\S]*?)\"";
		Pattern p = Pattern.compile(regex);
		Matcher m = p.matcher(sqlCol);
		while (m.find()) {
			String key = m.group().replaceAll("\"", "");
			JSONObject col = new JSONObject();
			col.put("name", key.replaceAll("\\s*", ""));
			col.put("value", "");
			after.put(col);
		}
		// 截取列值字符串
		String sqlVal = sql.substring(endColIndex, sql.length());
		sqlVal = sqlVal.substring(sqlVal.indexOf("(") + 1, sqlVal.lastIndexOf(")"));
		sqlVal = "SELECT " + sqlVal + " FROM DUAL";
		List<Object> vals = getValsFromSql(jdbcTemplate, sqlVal);
		int aLen = after.length();
		for (int i = 0; i < aLen; i++) {
			JSONObject col = (JSONObject) after.get(i);
			Object val = vals.get(i);
			val = val == null ? "" : val;
			col.put("value", val);
		}
		insert.put("eventType", oper);
		insert.put("before", before);
		insert.put("after", after);
		return insert;
	}

	private static JSONObject update(JdbcTemplate jdbcTemplate, String oper, String sql) throws JSONException {
		// 模拟修改消息
		JSONObject update = new JSONObject();
		// update "YANGP"."ASD_USER" set "NAME" = '669' where "ID" = '6' and
		// "NAME" = '66922' and ROWID = 'AAAR75AAEAAAAC3AAC';

		// 截取表
		String scanTable = sql.substring(sql.indexOf("update") + 6, sql.indexOf("set"));

		// 为了提高性能，去掉变化前的数据，直接获取最新的数据
		// 解析ROWID,查询该条数据所有列名及值
		String sqlCol = sql.substring(sql.lastIndexOf("and") + 3, sql.lastIndexOf(";"));
		JSONArray after = updateAfter(jdbcTemplate, scanTable, sqlCol);

		update.put("eventType", oper);
		update.put("before", new JSONArray());
		update.put("after", after);
		return update;
	}

	private static JSONArray updateAfter(JdbcTemplate jdbcTemplate, String scanTable, String sqlCol)
			throws JSONException {
		JSONArray before = new JSONArray();
		StringBuilder sql = new StringBuilder("SELECT * FROM ").append(scanTable).append(" WHERE " + sqlCol);

		// 解析列名
		SqlRowSet rs = jdbcTemplate.queryForRowSet(sql + "");
		SqlRowSetMetaData sqlRsmd = rs.getMetaData();
		int columnCount = sqlRsmd.getColumnCount();
		for (int i = 1; i <= columnCount; i++) {
			JSONObject col = new JSONObject();
			col.put("name", sqlRsmd.getColumnName(i).toUpperCase());
			col.put("value", "");
			before.put(col);
		}

		// 解析列值
		int columns = rs.getMetaData().getColumnCount();
		// 只有一行数据，调用一次next();
		rs.next();
		for (int i = 1; i <= columns; i++) {
			Object obj = rs.getObject(i);
			if (obj instanceof oracle.sql.TIMESTAMP) {
				try {
					obj = oracle.sql.TIMESTAMP.toTimestamp(((oracle.sql.TIMESTAMP) obj).getBytes());
				} catch (SQLException e) {
					e.printStackTrace();
				}
			}
			obj = obj == null ? "" : obj;
			before.getJSONObject(i - 1).put("value", obj);
		}
		return before;
	}

	private static JSONObject delete(JdbcTemplate jdbcTemplate, String oper, String sql) throws JSONException {
		JSONObject delete = new JSONObject();
		JSONArray before = new JSONArray();

		String sqlCol = sql.substring(sql.indexOf("where") + 5, sql.length() - 1);
		// 找出列名
		String regex = "\"([\\s\\S]*?)\"";
		Pattern p = Pattern.compile(regex);
		Matcher m = p.matcher(sqlCol);
		while (m.find()) {
			String key = m.group().replaceAll("\"", "");
			JSONObject col = new JSONObject();
			col.put("name", key.replaceAll("\\s*", ""));
			col.put("value", "");
			before.put(col);
		}

		// 清洗表达式
		sqlCol = sqlCol.replaceAll("IS NULL", "= NULL");
		// 找出值
		StringBuilder sqlVal = new StringBuilder("SELECT ");
		String[] colArr = sqlCol.split("and");
		int cLen = colArr.length;
		boolean isAppend = false;
		for (int i = 0; i < cLen; i++) {
			String[] attr = colArr[i].split("=");
			// 判断是否已经添加参数
			if (isAppend) {
				sqlVal.append(",");
			}
			sqlVal.append(attr[1]);
			isAppend = true;
		}
		sqlVal.append(" FROM DUAL");
		List<Object> vals = getValsFromSql(jdbcTemplate, sqlVal + "");
		int bLen = before.length();
		for (int i = 0; i < bLen; i++) {
			JSONObject col = (JSONObject) before.get(i);
			Object val = vals.get(i);
			val = val == null ? "" : val;
			col.put("value", val);
		}

		delete.put("eventType", oper);
		delete.put("before", before);
		delete.put("after", new JSONArray());
		return delete;
	}

	/**
	 * 转码
	 * 
	 * @param sql
	 * @return
	 */
	private static List<Object> getValsFromSql(JdbcTemplate jdbcTemplate, String sql) {
		List<Object> vals = new ArrayList<Object>();
		SqlRowSet rs = jdbcTemplate.queryForRowSet(sql);
		int columns = rs.getMetaData().getColumnCount();
		while (rs.next()) {
			for (int i = 1; i <= columns; i++) {
				Object obj = rs.getObject(i);
				if (obj instanceof oracle.sql.TIMESTAMP) {
					try {
						obj = oracle.sql.TIMESTAMP.toTimestamp(((oracle.sql.TIMESTAMP) obj).getBytes());
					} catch (SQLException e) {
						e.printStackTrace();
					}
				}
				vals.add(obj);
			}
		}
		return vals;
	}

}
